#include "SubStreamServer.h"
#include "EventLoop.h"
#include "Manager.h"
#include "PlatLogger.h"
#include "StreamSocketChannel.h"
#include "Util.h"
#include "XnccAddress.h"
#include "XnccType.h"

namespace xncc {
namespace foundation {

    SubStreamServer::SubStreamServer(EventLoop* loop) : loop_(loop)
    {
        if (loop_ == nullptr) {
            std::ostringstream oss;
            oss << " SubStreamServer engine must have a valid eventloop";
            throw shared::logic_exception(oss.str());
        }

        loop_->registerUserFunc(shared::InnerMsgId::ON_PUB_IPC_MSG,
                                [this](const char* msg, uint32_t len, types::thread_id_t tid,
                                       types::connection_id_t connId) { handleIoMsg(msg, len, tid, connId); });
    }

    SubStreamServer::~SubStreamServer() = default;

    void SubStreamServer::acceptNewConnection(int sockfd, std::shared_ptr<XnccAddress> peerAddr)
    {
        loop_->runInLoop([this, sockfd, peerAddr] { newConnection(sockfd, peerAddr); });
    }

    void SubStreamServer::handleIoMsg(const char* msg,
                                      uint32_t    len,
                                      types::thread_id_t /*tid*/,
                                      types::connection_id_t connId)
    {
        auto ptr = clientChs_[connId];
        if (ptr == nullptr) {
            TRACE_ERROR << "msg sender set error connid:"_s << connId;
            return;
        }
        ptr->sendMsg(msg, len);
    }

    void SubStreamServer::newConnection(int sockfd, std::shared_ptr<XnccAddress> peerAddr)
    {
        const auto connId = shared::CommonBusi::generateConnId();
        // 构建消息
        const std::string peerAddrStr = peerAddr->toStr();
        QUEUE_MGR.pubIPCDataToBusiThread(peerAddrStr.c_str(), static_cast<uint32_t>(peerAddrStr.size()),
                                         shared::InnerMsgId::ON_CONNECTED, loop_->id(), connId);
        TRACE_INFO << " engine id:"_s << loop_->id() << " new connection, connid:"_s << connId << " peeraddr:"_s
                   << peerAddrStr;
        // 一个连接建立之后，投递连接信息到 关注消息的线程， 然后要把这个fd，放进epool循环中
        /// 这里创建的channel
        /// 是TCP连接的channel，断开连接时需要释放内存的，使用智能指针，但是连接断开的时候，只要erase了这个指针，内存才能被释放
        auto chPtr         = std::make_shared<StreamSocketChannel>(loop_, sockfd, connId);
        clientChs_[connId] = chPtr;
        chPtr->init();
        chPtr->setCloseCallback([this, connId](const std::string& reason) { removeConnection(connId, reason); });
        chPtr->enableReading();
    }

    void SubStreamServer::removeConnection(types::connection_id_t connId, const std::string& disconnectedReason)
    {
        auto iter = clientChs_.find(connId);
        if (iter != clientChs_.end()) {
            TRACE_INFO << " engine id:"_s << loop_->id() << " connection disconnected, connid:"_s << connId;
            QUEUE_MGR.pubIPCDataToBusiThread(disconnectedReason.c_str(),
                                             static_cast<uint32_t>(disconnectedReason.size()),
                                             shared::InnerMsgId::ON_DISCONNECTED, loop_->id(), connId);
            iter->second->disableAll();
            iter->second->remove();
            clientChs_.erase(connId);
        }
        else {
            TRACE_INFO << "client disconnect but not founnd connId, fd = "_s << connId;
        }
    }
}  // namespace foundation
}  // namespace xncc